package com.example.client.netty;

import com.example.client.ClientImpl;
import com.example.transport.message.*;
import com.example.transport.protocol.Transport;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.nio.NioSocketChannel;
import io.netty.handler.codec.serialization.ClassResolver;
import io.netty.handler.codec.serialization.ClassResolvers;
import io.netty.handler.codec.serialization.ObjectDecoder;
import io.netty.handler.codec.serialization.ObjectEncoder;
import io.netty.handler.timeout.IdleState;
import io.netty.handler.timeout.IdleStateEvent;
import io.netty.handler.timeout.IdleStateHandler;

import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;
import java.util.logging.Logger;

/****
 *
 * 默认netty的客户端
 */
public class DefaultNettyClient extends ClientImpl implements Transport {
    Logger logger = Logger.getLogger(getClass().getName());
    private Map<String, ResponseFuture> futureMap = new HashMap<>();

    private NioEventLoopGroup nioEventLoopGroup;

    private Channel channel;

    public DefaultNettyClient(String ip, int port) {
        super(ip, port);
    }

    @Override
    public void init() throws Exception {

    }

    @Override
    public void start() throws Exception {
        this.nioEventLoopGroup = new NioEventLoopGroup(10);
        Bootstrap bootstrap = new Bootstrap().group(nioEventLoopGroup).option(ChannelOption.TCP_NODELAY, true).channel(NioSocketChannel.class).handler(new ChannelInitializer() {
            @Override
            protected void initChannel(Channel ch) throws Exception {
                ClassResolver classResolver = ClassResolvers.cacheDisabled(getClass().getClassLoader());
                ch.pipeline().addLast("decoder", new ObjectDecoder(classResolver));
                ch.pipeline().addLast("encoder", new ObjectEncoder());
                ch.pipeline().addLast(new IdleStateHandler(5, 0, 0, TimeUnit.SECONDS) {
                    @Override
                    protected void channelIdle(ChannelHandlerContext ctx, IdleStateEvent evt) throws Exception {
                        if (evt.state() == IdleState.READER_IDLE) {
                            logger.warning("READER_IDLE READER_IDLE READER_IDLE ...");
                        }
                    }
                });
                ch.pipeline().addLast(new ChannelDuplexHandler() {
                    @Override
                    public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
                        channel = ctx.channel();
                        if (msg instanceof Response) {
                            String id = ((Response) msg).getRequest().getId();
                            if (futureMap.containsKey(id)) {
                                ResponseFuture responseFuture = futureMap.remove(id);
                                responseFuture.setResponse((Response) msg);
                            }
                        }
                    }
                });
            }
        });
        ChannelFuture channelFuture = bootstrap.connect(getIp(), getPort()).sync();
        if (channelFuture.isSuccess()) {
            channel = channelFuture.channel();
        }
    }

    @Override
    public void stop() throws Exception {
        if (!this.nioEventLoopGroup.isShutdown()) {
            this.nioEventLoopGroup.shutdownGracefully();
        }
    }

    @Override
    public void asyncSend(Request request) throws Exception {
        this.channel.writeAndFlush(request).sync();
    }

    @Override
    public Response syncSend(Request request) throws Exception {
        this.channel.writeAndFlush(request).sync();
        futureMap.put(request.getId(), new ResponseFuture(request.getId()));
        ResponseFuture responseFuture = futureMap.get(request.getId());
        Response rpcResponse = responseFuture.getRpcResponse();
        futureMap.remove(request.getId());
        return rpcResponse;
    }
}
